package cn.jly.flink;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.net.URL;

/**
 * @PackageName cn.jly.flink
 * @ClassName WordCount
 * @Description 入门案例word count
 * @Author 姬岚洋
 * @Date 2021/1/6 下午 10:00
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // source
        final URL resource = WordCount.class.getClassLoader().getResource("test/words.txt");
        assert resource != null;
        final DataSource<String> text = env.readTextFile(resource.getPath());

        // transform
        final AggregateOperator<Tuple2<String, Integer>> counts = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                final String[] words = s.split("\\W+");
                for (String word : words) {
                    if (StringUtils.isNotEmpty(word)) {
                        collector.collect(Tuple2.of(word, 1));
                    }
                }
            }
        }).groupBy(0).sum(1);

        // sink
        counts.print();

        // 对于离线批处理的算子，如：“count()”、“collect()”或“print()”等既有sink功能，还有触发的功能。
        // 我们上面调用了print()方法，会自动触发execute，所以最后面的一行执行器没有数据可以执行
        // env.execute("word count");
    }
}
